package com.shujia.spark.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo2UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")

    val sc = new SparkContext(conf)

    /**
      * 创建spark streaming环境
      *
      */

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    //设置checkpoint的路径
    ssc.checkpoint("data/checkpoint")


    /**
      * 启动socket
      * yum install nc
      * nc -lk 8888
      *
      */

    //1、读取数据
    val lineDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)


    val wordDS: DStream[String] = lineDS.flatMap(line => line.split(","))

    val kvDS: DStream[(String, Int)] = wordDS.map((_, 1))


    /**
      * 更新状态的函数（单词的数量）
      *
      * @param seq    ： 当前批次每一个key所有的value
      * @param option : 之前计算的状态， 默认会保存到checkpoint中，所以需要指定checkpoint的路径
      * @return 返回最新的状态
      */
    def updataFun(seq: Seq[Int], option: Option[Int]): Option[Int] = {
      //获取之前的状态
      //之前单词的数量
      val lastCount: Int = option.getOrElse(0)

      //计算当前批处单词的数量
      val currCount: Int = seq.sum

      //计算最新单词的数量
      val count: Int = lastCount + currCount

      //返回最新单词的数量
      Some(count)
    }

    /**
      * updateStateByKey: 有状态算子
      * 状态; 单词的数量
      * 没批次使用当成批处的结果加上之前单词的结果更新总的单词的数量
      *
      */


    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(updataFun)


    //打印出来
    countDS.print()


    //启动spark streaming程序
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
